package com.atguigu.gmallrealtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.atguigu.gmallrealtime.app.func.DimMapFunctionHbaseAsync;
import com.atguigu.gmallrealtime.bean.TradeSkuOrderBean;
import com.atguigu.gmallrealtime.common.Constant;
import com.atguigu.gmallrealtime.util.DateFormatUtil;
import com.atguigu.gmallrealtime.util.DorisUtil;
import com.atguigu.gmallrealtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * @author yhm
 * @create 2023-10-12 9:32
 */
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1 创建flink环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 2 添加检查点和状态后端
        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);

        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置job取消之后 检查点是否保留
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置两个检查点之间最小的时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.seconds(3)));

        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");

        System.setProperty("HADOOP_USER_NAME", "atguigu");

        // TODO 3 读取dwd_trade_order_detail主题数据
        String groupId = "dws_trade_sku_order_window";
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, groupId);
        DataStreamSource<String> odSource = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "od_source");

        // TODO 4 过滤掉回撤流的null值 之后转换结构jsonObject
        SingleOutputStreamOperator<JSONObject> jsonObjStream = odSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    if (value != null) {
                        out.collect(JSON.parseObject(value));
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        // TODO 5 使用唯一键 od.id 进行分组去重
        KeyedStream<JSONObject, String> keyedStream = jsonObjStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getString("id");
            }
        });

        // TODO 6 使用状态将重复数据的度量值变为负数  转换为javaBean
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStream = keyedStream.map(new RichMapFunction<JSONObject, TradeSkuOrderBean>() {

            ValueState<JSONObject> lastOdState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                // 获取状态
                // 如果状态为null  说明当前为join过来的第一条数据
                // 如果状态不为null 说明需要回撤上一条数据
                ValueStateDescriptor<JSONObject> lastOdDes = new ValueStateDescriptor<>("last_od", JSONObject.class);
                lastOdDes.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(15L)).build());
                lastOdState = getRuntimeContext().getState(lastOdDes);
            }

            @Override
            public TradeSkuOrderBean map(JSONObject value) throws Exception {
                JSONObject lastOd = lastOdState.value();
                BigDecimal orderPrice = value.getBigDecimal("order_price");
                BigDecimal skuNum = value.getBigDecimal("sku_num");
                BigDecimal curOdOriginalAmount = orderPrice.multiply(skuNum);
                TradeSkuOrderBean skuOrderBean = TradeSkuOrderBean.builder()
                        .skuId(value.getString("sku_id"))
                        .skuName(value.getString("sku_name"))
                        .originalAmount(curOdOriginalAmount)
                        .activityAmount(value.getString("split_activity_amount") == null ? new BigDecimal("0.0") : value.getBigDecimal("split_activity_amount"))
                        .couponAmount(value.getString("split_coupon_amount") == null ? new BigDecimal("0.0") : value.getBigDecimal("split_coupon_amount"))
                        .orderAmount(value.getBigDecimal("split_total_amount"))
                        .ts(value.getLong("ts") * 1000)
                        .build();
                // 根据状态判断当前是否需要回撤上一条数据
                if (lastOd == null) {
                    // 当前为第一条数据
                    lastOdState.update(value);
                    return skuOrderBean;
                }
                // 当前不是第一条数据
                // 将上一条数据的度量值改为负数进行累加
                lastOdState.update(value);
                BigDecimal lastOrderPrice = lastOd.getBigDecimal("order_price");
                BigDecimal lastSkuNum = lastOd.getBigDecimal("sku_num");
                BigDecimal lastCurOdOriginalAmount = lastOrderPrice.multiply(lastSkuNum);
                skuOrderBean.setOriginalAmount(skuOrderBean.getOriginalAmount().subtract(lastCurOdOriginalAmount));
                skuOrderBean.setOrderAmount(skuOrderBean.getOrderAmount().subtract(lastOd.getBigDecimal("split_total_amount")));
                skuOrderBean.setCouponAmount(skuOrderBean.getCouponAmount().subtract(lastOd.getString("split_coupon_amount") == null ? new BigDecimal("0.0") : lastOd.getBigDecimal("split_coupon_amount")));
                skuOrderBean.setActivityAmount(skuOrderBean.getActivityAmount().subtract(lastOd.getString("split_activity_amount") == null ? new BigDecimal("0.0") : lastOd.getBigDecimal("split_activity_amount")));
                return skuOrderBean;
            }
        });


        // TODO 7 添加水位线
        SingleOutputStreamOperator<TradeSkuOrderBean> withWaterMarkStream = beanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeSkuOrderBean>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<TradeSkuOrderBean>() {
            @Override
            public long extractTimestamp(TradeSkuOrderBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));

        // TODO 8 按照sku_id分组
        KeyedStream<TradeSkuOrderBean, String> skuIdStream = withWaterMarkStream.keyBy(new KeySelector<TradeSkuOrderBean, String>() {
            @Override
            public String getKey(TradeSkuOrderBean value) throws Exception {
                return value.getSkuId();
            }
        });

        // TODO 9 开窗聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> windowStream = skuIdStream.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10))).allowedLateness(org.apache.flink.streaming.api.windowing.time.Time.seconds(5L)).reduce(new ReduceFunction<TradeSkuOrderBean>() {
            @Override
            public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                return value1;
            }
        }, new ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
            @Override
            public void process(String s, Context context, Iterable<TradeSkuOrderBean> elements, Collector<TradeSkuOrderBean> out) throws Exception {
                TimeWindow window = context.window();
                String stt = DateFormatUtil.toYmdHms(window.getStart());
                String edt = DateFormatUtil.toYmdHms(window.getEnd());
                String curDt = DateFormatUtil.toDorisDate(DateFormatUtil.toTs(stt, true));
                for (TradeSkuOrderBean element : elements) {
                    element.setStt(stt);
                    element.setEdt(edt);
                    element.setCurDate(curDt);
                    out.collect(element);
                }
            }
        });

//        windowStream.print();

        // TODO 10 维度关联 补全维度信息
        // 10.1 维度关联sku_info 补全tmId和c3Id
//        SingleOutputStreamOperator<TradeSkuOrderBean> skuBean = windowStream.map(new DimMapFunctionHbase<TradeSkuOrderBean>() {
//            @Override
//            public String getTable() throws Exception {
//                return "dim_sku_info";
//            }
//
//            @Override
//            public String getId(TradeSkuOrderBean value) throws Exception {
//                return value.getSkuId();
//            }
//
//            @Override
//            public void join(TradeSkuOrderBean value, JSONObject dim) throws Exception {
//                value.setCategory3Id(dim.getString("category3_id"));
//                value.setTrademarkId(dim.getString("tm_id"));
//                value.setSpuId(dim.getString("spu_id"));
//            }
//        });
//
//        SingleOutputStreamOperator<TradeSkuOrderBean> spuBean = skuBean.map(new DimMapFunctionHbase<TradeSkuOrderBean>() {
//            @Override
//            public String getTable() throws Exception {
//                return "dim_spu_info";
//            }
//
//            @Override
//            public String getId(TradeSkuOrderBean value) throws Exception {
//                return value.getSpuId();
//            }
//
//            @Override
//            public void join(TradeSkuOrderBean value, JSONObject dim) throws Exception {
//                value.setSpuName(dim.getString("spu_name"));
//            }
//        });
//
//
//        // 10.2 维度关联c3表 关联c3name和c2Id
//        SingleOutputStreamOperator<TradeSkuOrderBean> c3Bean = spuBean.map(new DimMapFunctionHbase<TradeSkuOrderBean>() {
//            @Override
//            public String getTable() throws Exception {
//                return "dim_base_category3";
//            }
//
//            @Override
//            public String getId(TradeSkuOrderBean value) throws Exception {
//                return value.getCategory3Id();
//            }
//
//            @Override
//            public void join(TradeSkuOrderBean value, JSONObject dim) throws Exception {
//                value.setCategory3Name(dim.getString("name"));
//                value.setCategory2Id(dim.getString("category2_id"));
//            }
//        });
//
//        // 10.3 维度关联c2表 关联c2name和c1id
//        SingleOutputStreamOperator<TradeSkuOrderBean> c2bean = c3Bean.map(new DimMapFunctionHbase<TradeSkuOrderBean>() {
//            @Override
//            public String getTable() throws Exception {
//                return "dim_base_category2";
//            }
//
//            @Override
//            public String getId(TradeSkuOrderBean value) throws Exception {
//                return value.getCategory2Id();
//            }
//
//            @Override
//            public void join(TradeSkuOrderBean value, JSONObject dim) throws Exception {
//                value.setCategory2Name(dim.getString("name"));
//                value.setCategory1Id(dim.getString("category1_id"));
//            }
//        });
//
//        // 10.4 维度关联c1表 关联c1name
//        SingleOutputStreamOperator<TradeSkuOrderBean> c1Bean = c2bean.map(new DimMapFunctionHbase<TradeSkuOrderBean>() {
//            @Override
//            public String getTable() throws Exception {
//                return "dim_base_category1";
//            }
//
//            @Override
//            public String getId(TradeSkuOrderBean value) throws Exception {
//                return value.getCategory1Id();
//            }
//
//            @Override
//            public void join(TradeSkuOrderBean value, JSONObject dim) throws Exception {
//                value.setCategory1Name(dim.getString("name"));
//            }
//        });
//
//        // 10.5 维度关联tm表 关联tmName
//        SingleOutputStreamOperator<TradeSkuOrderBean> dimBean = c1Bean.map(new DimMapFunctionHbase<TradeSkuOrderBean>() {
//            @Override
//            public String getTable() throws Exception {
//                return "dim_base_trademark";
//            }
//
//            @Override
//            public String getId(TradeSkuOrderBean value) throws Exception {
//                return value.getTrademarkId();
//            }
//
//            @Override
//            public void join(TradeSkuOrderBean value, JSONObject dim) throws Exception {
//                value.setTrademarkName(dim.getString("tm_name"));
//            }
//        });

        // 使用flink的异步算子来实现维度关联
        // 关联sku_info
        SingleOutputStreamOperator<TradeSkuOrderBean> skuBean = AsyncDataStream.unorderedWait(windowStream, new DimMapFunctionHbaseAsync<TradeSkuOrderBean>() {
            @Override
            public String getTable() throws Exception {
                return "dim_sku_info";
            }

            @Override
            public String getId(TradeSkuOrderBean value) throws Exception {
                return value.getSkuId();
            }

            @Override
            public void join(TradeSkuOrderBean value, JSONObject dim) throws Exception {
                value.setCategory3Id(dim.getString("category3_id"));
                value.setTrademarkId(dim.getString("tm_id"));
                value.setSpuId(dim.getString("spu_id"));
            }
        }, 60L, TimeUnit.SECONDS);

        // 关联spu_info
        SingleOutputStreamOperator<TradeSkuOrderBean> spuBean = AsyncDataStream.unorderedWait(skuBean, new DimMapFunctionHbaseAsync<TradeSkuOrderBean>() {
            @Override
            public String getTable() throws Exception {
                return "dim_spu_info";
            }

            @Override
            public String getId(TradeSkuOrderBean value) throws Exception {
                return value.getSpuId();
            }

            @Override
            public void join(TradeSkuOrderBean value, JSONObject dim) throws Exception {
                value.setSpuName(dim.getString("spu_name"));
            }
        }, 60L, TimeUnit.SECONDS);

        // 关联c3表
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Bean = AsyncDataStream.unorderedWait(spuBean, new DimMapFunctionHbaseAsync<TradeSkuOrderBean>() {
            @Override
            public String getTable() throws Exception {
                return "dim_base_category3";
            }

            @Override
            public String getId(TradeSkuOrderBean value) throws Exception {
                return value.getCategory3Id();
            }

            @Override
            public void join(TradeSkuOrderBean value, JSONObject dim) throws Exception {
                value.setCategory3Name(dim.getString("name"));
                value.setCategory2Id(dim.getString("category2_id"));
            }
        }, 60L, TimeUnit.SECONDS);

        // 关联c2
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Bean = AsyncDataStream.unorderedWait(c3Bean, new DimMapFunctionHbaseAsync<TradeSkuOrderBean>() {
            @Override
            public String getTable() throws Exception {
                return "dim_base_category2";
            }

            @Override
            public String getId(TradeSkuOrderBean value) throws Exception {
                return value.getCategory2Id();
            }

            @Override
            public void join(TradeSkuOrderBean value, JSONObject dim) throws Exception {
                value.setCategory2Name(dim.getString("name"));
                value.setCategory1Id(dim.getString("category1_id"));
            }
        }, 60L, TimeUnit.SECONDS);

        // 关联c1表
        SingleOutputStreamOperator<TradeSkuOrderBean> c1Bean = AsyncDataStream.unorderedWait(c2Bean, new DimMapFunctionHbaseAsync<TradeSkuOrderBean>() {
            @Override
            public String getTable() throws Exception {
                return "dim_base_category1";
            }

            @Override
            public String getId(TradeSkuOrderBean value) throws Exception {
                return value.getCategory1Id();
            }

            @Override
            public void join(TradeSkuOrderBean value, JSONObject dim) throws Exception {
                value.setCategory1Name(dim.getString("name"));
            }
        }, 60L, TimeUnit.SECONDS);

        // 关联tm表
        SingleOutputStreamOperator<TradeSkuOrderBean> dimBean = AsyncDataStream.unorderedWait(c1Bean, new DimMapFunctionHbaseAsync<TradeSkuOrderBean>() {
            @Override
            public String getTable() throws Exception {
                return "dim_base_trademark";
            }

            @Override
            public String getId(TradeSkuOrderBean value) throws Exception {
                return value.getTrademarkId();
            }

            @Override
            public void join(TradeSkuOrderBean value, JSONObject dim) throws Exception {
                value.setTrademarkName(dim.getString("tm_name"));
            }
        }, 60L, TimeUnit.SECONDS);

//        dimBean.print();

        // TODO 11 写出到doris中
        dimBean.map(new MapFunction<TradeSkuOrderBean, String>() {
            @Override
            public String map(TradeSkuOrderBean value) throws Exception {
                SerializeConfig config = new SerializeConfig();
                config.setPropertyNamingStrategy(PropertyNamingStrategy.SnakeCase);
                return JSON.toJSONString(value, config);
            }
        })
                .sinkTo(DorisUtil.getDorisSink("dws_trade_sku_order_window"));

        // TODO 12 执行任务
        env.execute();

    }
}
